package com.weimob.mcloud.wechatwork.customer.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
import com.weimob.artemis.client.consumer.model.ConsumerRecord;
import com.weimob.artemis.client.consumer.model.MessageResult;
import com.weimob.artemis.client.spring.annotation.ArtemisConsumerListener;
import com.weimob.artemis.client.spring.annotation.Subscribe;
import com.weimob.mcloud.wechatwork.customer.domain.ArtemisConstant;
import com.weimob.mcloud.wechatwork.customer.domain.EmployeeAuthorizeChangeInfo;
import com.weimob.mcloud.wechatwork.customer.mq.handler.employee.EmployeeAuthorizeChangeForGroupHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * 员工授权范围变更消息消费
 */
@Component
@Slf4j
public class EmployeeAuthorizeChangeForGroupConsumer {

    @Autowired
    @Qualifier("employeeAuthorizeChangeForGroupThreadPool")
    private ThreadPoolTaskExecutor employeeAuthorizeChangeForGroupThreadPool;

    @Autowired
    @Qualifier("employeeSeatChangeForGroupThreadPool")
    private ThreadPoolTaskExecutor employeeSeatChangeForGroupThreadPool;

    @Autowired
    private EmployeeAuthorizeChangeForGroupHandler employeeAuthorizeChangeForGroupHandler;

    @ApolloJsonValue("${employee.statusChange.blackList.group.corpid:[]}")
    private List<String> employeeStatusChangeBlackListGroupCorpId;

    /**
     * 员工授权范围变更消息消费for客户群
     */
    @ArtemisConsumerListener(
            value = @Subscribe(topic = ArtemisConstant.Topic.WECHATWORK_EMPLOYEE_DEP_AUTHORIZE_MSG, user = ArtemisConstant.CONSUMER_USER_FOR_GROUP))
    public void employeeAuthorizeChangeConsumer(MessageResult result) {
        List<ConsumerRecord> records = result.getRecords();

        CompletableFuture<Void>[] completableFutures = new CompletableFuture[records.size()];
        for (int i = 0; i < records.size(); i++) {

            ConsumerRecord consumerRecord = records.get(i);
            completableFutures[i] = CompletableFuture.runAsync(() -> {
                log.info("客户群-消费员工授权范围变更 message：{}", JSON.toJSONString(consumerRecord));

                try {
                    EmployeeAuthorizeChangeInfo employeeAuthorizeChangeInfo = JSONObject.parseObject(consumerRecord.getValue(), EmployeeAuthorizeChangeInfo.class);
                    if (CollectionUtils.isNotEmpty(employeeStatusChangeBlackListGroupCorpId) && employeeStatusChangeBlackListGroupCorpId.contains(employeeAuthorizeChangeInfo.getCorpId())) {
                        log.info("客户群-黑名单拦截消费员工授权范围变更 message：{}", JSON.toJSONString(employeeAuthorizeChangeInfo));
                        return;
                    }

                    employeeAuthorizeChangeForGroupHandler.handleEmployeeAuthorizeChange(employeeAuthorizeChangeInfo);
                } catch (Exception e) {
                    log.error("客户群-消费员工授权范围变更 Artemis数据处理失败. error:{}", e);
                }

            }, employeeAuthorizeChangeForGroupThreadPool);

        }
        // 等待执行完成
        CompletableFuture.allOf(completableFutures).join();
    }


    /**
     * 员工坐席限制变更消息消费for客户群
     */
    @ArtemisConsumerListener(
            value = @Subscribe(topic = ArtemisConstant.Topic.WECHATWORK_EMPLOYEE_SEAT_AUTH_MSG, user = ArtemisConstant.CONSUMER_USER_FOR_GROUP), pollSize = 1)
    public void employeeSeatAuthChangeConsumer(MessageResult result) {
        List<ConsumerRecord> records = result.getRecords();

        CompletableFuture<Void>[] completableFutures = new CompletableFuture[records.size()];
        for (int i = 0; i < records.size(); i++) {

            ConsumerRecord consumerRecord = records.get(i);
            completableFutures[i] = CompletableFuture.runAsync(() -> {
                log.info("客户群-消费员工坐席变更 message：{}", JSON.toJSONString(consumerRecord));

                try {
                    EmployeeAuthorizeChangeInfo employeeAuthorizeChangeInfo = JSONObject.parseObject(consumerRecord.getValue(), EmployeeAuthorizeChangeInfo.class);
                    if (null == employeeAuthorizeChangeInfo) {
                        return;
                    }
                    if (CollectionUtils.isNotEmpty(employeeStatusChangeBlackListGroupCorpId) && employeeStatusChangeBlackListGroupCorpId.contains(employeeAuthorizeChangeInfo.getCorpId())) {
                        log.info("客户群-黑名单拦截消费员工授权范围变更 message：{}", JSON.toJSONString(employeeAuthorizeChangeInfo));
                        return;
                    }
                    employeeAuthorizeChangeForGroupHandler.handleEmployeeAuthorizeChange(employeeAuthorizeChangeInfo);
                } catch (Exception e) {
                    log.error("客户群-消费员工坐席变更 Artemis数据处理失败. error", e);
                }

            }, employeeSeatChangeForGroupThreadPool);

        }
        // 等待执行完成
        CompletableFuture.allOf(completableFutures).join();

    }
}
